package org.zjt.rxjava.lession.second;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 *
 *      1、doOn .....  作为观察者模式，在状态响应事件
 *      2、合并
 *          2.1、merge 多个 Observable
 *          2.2、mergeWith
 *      3、flatMap
 *      4、连接
 *          4.1、concat
 *          4.2、concatWith
 *
 *      5、amb 只发送第一个抢占到锁的发布者
 *
 *      6、zip 、zipWith
 *
 *      7、combineLatest 最新的合并
 *
 *
 *
 *
 * @author juntao.zhang
 * Date: 2018-06-28 下午5:21
 * @see
 */
@Slf4j
public class ActionOperatorsTest {

    public static void main(String[] args) throws InterruptedException {

        Observable.just("Alpha", "Beta", "Gamma", "Delta", "Epsilon")
                .doOnNext(s -> log.info("Processing: " + s))
                .doAfterNext(s -> log.info("Processing: " + s))
                .doOnSubscribe(s -> log.info("Processing: " + s))
                .doOnComplete(() -> log.info("complete"))
                .doOnError(e -> log.error("异常",e))
                .onErrorReturn(e -> "00000")
                .doOnDispose(() -> log.info("----------Dispose----------") )
                .subscribeOn(Schedulers.io())
                .map(String::length)
                .blockingSubscribe(i -> log.info("Received: " + i));


        Observable.just(5, 3, 7, 10, 2, 14)
                .reduce((total, next) -> total + next)
                .doOnSuccess(i -> log.info("Emitting: " + i))
                .subscribe(i -> log.info("Received: " + i));



        Observable<String> source1 = Observable.just("Alpha", "Beta", "Gamma", "Delta", "Epsilon");
        Observable<String> source2 = Observable.just("Zeta", "Eta", "Theta");
        Observable.merge(source1, source2).subscribe(i -> System.out.println("RECEIVED: " + i));


        Observable<String> source = Observable.just("Alpha", "Beta", "Gamma", "Delta", "Epsilon");
        source.flatMap(s -> Observable.fromArray(s.split(""))).subscribe(System.out::println);






        //emit every second
        Observable<String> source3 = Observable.interval(1, TimeUnit.SECONDS)
                        .map(l -> l + 1) // emit elapsed seconds
                        .map(l -> "Source1: " + l + " seconds");
        //emit every 300 milliseconds
        Observable<String> source4 = Observable.interval(300, TimeUnit.MILLISECONDS)
                .map(l -> (l + 1) * 300) // emit elapsed
                .map(l -> "Source2: " + l + " milliseconds");
        //emit Observable that emits first
        Observable.amb(Arrays.asList(source3, source4)).subscribe(i -> System.out.println("RECEIVED  22: " +i));
        //keep application alive for 5 seconds


        TimeUnit.SECONDS.sleep(5);
                
    }
}




